package caozuoxitong;

import java.util.concurrent.Semaphore;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;


public class ProducerConsumerSemaphore {

    // 共享资源定义
    private static final int BUFFER_SIZE = 5;        // 缓冲区容量
    private static final int PRODUCER_NUM = 2;       // 生产者线程数
    private static final int CONSUMER_NUM = 3;       // 消费者线程数

    // 循环缓冲区
    private static int[] buffer = new int[BUFFER_SIZE];
    private static AtomicInteger in = new AtomicInteger(0);  // 写入指针
    private static AtomicInteger out = new AtomicInteger(0); // 读取指针

    // 信号量定义
    private static Semaphore empty;  // 记录空位数量
    private static Semaphore full;   // 记录数据数量
    private static Semaphore mutex;  // 互斥信号量

    // 线程控制标志
    private static volatile boolean running = true;

    /**
     * 生产者线程类
     */
    public static class Producer implements Runnable {
        private int producerId;
        private Random random = new Random();

        public Producer(int id) {
            this.producerId = id;
        }

        @Override
        public void run() {
            System.out.println("生产者 " + producerId + " 启动");

            while (running) {
                try {
                    // 1. 模拟生产数据(1~100的随机数)
                    int data = random.nextInt(100) + 1;
                    System.out.println("生产者 " + producerId + " 生成数据: " + data);

                    // 2. 申请空位(同步控制)
                    empty.acquire();

                    // 3. 申请临界区访问(互斥控制)
                    mutex.acquire();

                    // 4. 写入数据到缓冲区
                    int writePos = in.get();
                    buffer[writePos] = data;
                    System.out.println("生产者 " + producerId + " 写入缓冲区位置 " + writePos + ": " + data);

                    // 更新写入指针(循环)
                    in.set((writePos + 1) % BUFFER_SIZE);

                    // 5. 释放临界区
                    mutex.release();

                    // 6. 通知消费者"有新数据"
                    full.release();

                    // 模拟生产耗时(1~3秒)
                    Thread.sleep(random.nextInt(2000) + 1000);

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            System.out.println("生产者 " + producerId + " 结束");
        }
    }


     //消费者线程类

    public static class Consumer implements Runnable {
        private int consumerId;
        private Random random = new Random();

        public Consumer(int id) {
            this.consumerId = id;
        }

        @Override
        public void run() {
            System.out.println("消费者 " + consumerId + " 启动");

            while (running) {
                try {
                    // 1. 申请数据(同步控制)
                    full.acquire();

                    // 2. 申请临界区访问(互斥控制)
                    mutex.acquire();

                    // 3. 从缓冲区读取数据
                    int readPos = out.get();
                    int data = buffer[readPos];
                    System.out.println("消费者 " + consumerId + " 从缓冲区位置 " + readPos + " 读取数据: " + data);

                    // 更新读取指针(循环)
                    out.set((readPos + 1) % BUFFER_SIZE);

                    // 4. 释放临界区
                    mutex.release();

                    // 5. 通知生产者"有空位"
                    empty.release();

                    // 模拟消费耗时(1~3秒)
                    Thread.sleep(random.nextInt(3000) + 1000);

                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            System.out.println("消费者 " + consumerId + " 结束");
        }
    }

     //主函数 - 运行实验

    public static void main(String[] args) {
        System.out.println("基于信号量的生产者-消费者实验");
        System.out.println("缓冲区大小: " + BUFFER_SIZE);
        System.out.println("生产者数量: " + PRODUCER_NUM + ", 消费者数量: " + CONSUMER_NUM);

        runWithSemaphore();
    }

    private static void runWithSemaphore() {

        // 初始化信号量
        empty = new Semaphore(BUFFER_SIZE);  // 初始空位数为缓冲区大小
        full = new Semaphore(0);             // 初始数据数为0
        mutex = new Semaphore(1);            // 互斥信号量，初始为1

        // 重置指针
        in.set(0);
        out.set(0);

        Thread[] producers = new Thread[PRODUCER_NUM];
        Thread[] consumers = new Thread[CONSUMER_NUM];

        // 创建并启动生产者线程
        for (int i = 0; i < PRODUCER_NUM; i++) {
            producers[i] = new Thread(new Producer(i + 1));
            producers[i].start();
        }

        // 创建并启动消费者线程
        for (int i = 0; i < CONSUMER_NUM; i++) {
            consumers[i] = new Thread(new Consumer(i + 1));
            consumers[i].start();
        }

        // 运行10秒后停止
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        running = false;

        // 等待所有线程结束
        for (Thread producer : producers) {
            producer.interrupt();
        }
        for (Thread consumer : consumers) {
            consumer.interrupt();
        }

        System.out.println("实验结束\n");
    }

}
